package org.one.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.NoArgsConstructor;
import org.one.entity.TranslatorDataWrapper;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
 * @Classname : RingBufferWorkerPoolFactory
 * @Description : 多消费者工作队列工厂
 * @Author : cty
 * @Date 2023/6/11 20:52
 */

@NoArgsConstructor
public class RingBufferWorkerPoolFactory {

    /**
     * 单例队列
     */
    private static class SingletonHolder {
        static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
    }

    /**
     * 获取单例队列
     * @return
     */
    public static RingBufferWorkerPoolFactory getInstance() {
        return SingletonHolder.instance;
    }


    /**
     * 生产者池
     */
    private static Map<String, MessageProducer> producers = new ConcurrentHashMap<String, MessageProducer>();

    /**
     * 消费者池
     */
    private static Map<String, MessageConsumer> consumers = new ConcurrentHashMap<String, MessageConsumer>();

    /**
     * 高性能队列
     */
    private RingBuffer<TranslatorDataWrapper> ringBuffer;

    /**
     * 协调屏障
     */
    private SequenceBarrier sequenceBarrier;

    /**
     * 工作池
     */
    private WorkerPool<TranslatorDataWrapper> workerPool;

    /**
     * 初始化高性能队列
     * @param type
     * @param bufferSize
     * @param waitStrategy
     * @param messageConsumers
     */
    public void initAndStart(ProducerType type, int bufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
        //1.构建ringBuffer对象
        this.ringBuffer = RingBuffer.create(type,
                new EventFactory<TranslatorDataWrapper>() {
                    public TranslatorDataWrapper newInstance() {
                        return new TranslatorDataWrapper();
                    }
                },
                bufferSize,
                waitStrategy);

        //2.设置序号栅栏
        this.sequenceBarrier = this.ringBuffer.newBarrier();

        //3.把构建的消费者放入消费者池中管理
        for(MessageConsumer mc : messageConsumers){
            this.consumers.put(mc.getConsumerId(), mc);
        }

        //4.设置工作池
        this.workerPool = new WorkerPool<>(this.ringBuffer,
                this.sequenceBarrier,
                new EventExceptionHandler(),
                messageConsumers);
        //5.添加工作序列
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        //6.启动工作池
        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2));
    }

    /**
     * 获取生产者
     * @param producerId
     * @return
     */
    public MessageProducer getMessageProducer(String producerId){
        //池里有直接获取生产者
        MessageProducer messageProducer = this.producers.get(producerId);
        if(null == messageProducer) {
            messageProducer = new MessageProducer(producerId, this.ringBuffer);
            this.producers.put(producerId, messageProducer);
        }
        return messageProducer;
    }

    /**
     * 异常静态类
     */
    static class EventExceptionHandler implements ExceptionHandler<TranslatorDataWrapper> {
        public void handleEventException(Throwable ex, long sequence, TranslatorDataWrapper event) {
        }
        public void handleOnStartException(Throwable ex) {
        }
        public void handleOnShutdownException(Throwable ex) {
        }
    }

}
